package com.wang.helloworld.hdfs;

import java.io.IOException;
import java.io.InputStream;
import java.io.PrintWriter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.security.UserGroupInformation;

/**
 * Hello world!
 *
 */
public class HelloHDFS 
{
	final static Configuration conf = new Configuration();
	
    public static void main( String[] args ) throws IOException
    {
    	HelloHDFS hdfs = new HelloHDFS();

    	//hdfs.list("/");

    	String path = "/test";
    	hdfs.createFolder(path);
    	hdfs.list(path);
    	hdfs.delete(path, true);
    	try{hdfs.list(path);}catch(Exception e){System.out.println(e.getMessage());}
    	
    	String filePath=path + "/" + "testFile.txt";
    	try{hdfs.createFile(filePath);}catch(Exception e){System.out.println(e.getMessage());}
    	hdfs.list(filePath);

    	hdfs.append(filePath, "abcdefg\n");
    	hdfs.append(filePath, "1234567\n");
    	
    	InputStream in = hdfs.read(filePath);
    	byte[] buffer = new byte[8];
    	while(0 != in.read(buffer))
    	{
    		System.out.println(new String(buffer, "UTF-8"));
    	}
    	in.close();
    	
    	hdfs.delete(filePath, true);
    	try{hdfs.list(filePath);}catch(Exception e){System.out.println(e.getMessage());}
    }
    
    public HelloHDFS()
    {
    	conf.addResource(new Path("core-site.xml"));
        conf.addResource(new Path("hdfs-site.xml"));
        conf.setBoolean(DFSConfigKeys.DFS_SUPPORT_APPEND_KEY, true);
    }
    
    protected static void authentication(Configuration conf)
    {
    	/**
    	 * Kerbro authentication:
    	 * 
    	 * conf.set("hadoop.security.authentication", "kerberos");
    	 * UserGroupInformation.setConfiguration(conf);
    	 * 
    	 * String principal = "hadoop/localhost@LOCALHOST";
    	 * String keytab = "xxx.keytab";
    	 * UserGroupInformation.loginUserFromKeytab(principal, keytab);
    	 */
    	
    	conf.set("hadoop.security.authentication", "simple");
    	UserGroupInformation.setConfiguration(conf);

    	UserGroupInformation ugi = UserGroupInformation.createRemoteUser("hadoop");
    	UserGroupInformation.setLoginUser(ugi);
    }
    
    public void list(String path) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);
    	
    	FileStatus[] status = fs.listStatus(new Path(path));
		for (int i = 0; i < status.length; i++) {
			System.out.print(status[i].getPath());
			if(status[i].isDirectory())
			{
				System.out.println(" [Dir]");
				list(status[i].getPath().toString());
			}
			else if(status[i].isFile())
			{
				System.out.println(" [File]");
				/*BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(status[i].getPath())));
				String line;
				try {
					line = br.readLine();
					while (line != null) {
						System.out.println(line);
						line = br.readLine();
					}
				} catch (Exception e) {
					System.out.println(e.getMessage());
				}*/
			}
			else if(status[i].isSymlink())
			{
				System.out.println(" [Link:"+ status[i].getSymlink() + "]");
			}
		}
		fs.close();
    }
    
    public void createFolder(String path) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);
    	
    	fs.mkdirs(new Path(path));
    	
    	fs.close();
    }
    
    public void delete(String path, boolean recursive) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);
    	
    	fs.delete(new Path(path), recursive);
    	
    	fs.close();
    }
    
    public void createFile(String filePath) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);
    	
    	fs.createNewFile(new Path(filePath));
    	
    	fs.close();
    }
    
    public void append(String filePath, String content) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);
    	boolean flag = Boolean.valueOf(fs.getConf().get("dfs.support.append"));
    	System.out.println("dfs.support.append is set to be " + flag);

    	FSDataOutputStream output = fs.append(new Path(filePath));
    	
    	PrintWriter writer = new PrintWriter(output);
    	writer.append(content);
    	writer.close();

    	fs.close();
    }
    
    public InputStream read(String filePath) throws IOException
    {
    	authentication(conf);
    	FileSystem fs = FileSystem.get(conf);

    	FSDataInputStream input = fs.open(new Path(filePath));

    	fs.close();
    	
    	return input.getWrappedStream();
    }
}
